package com.xy.kafka;

import org.apache.kafka.clients.admin.*;

import java.util.Arrays;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;

public class KafkaTopicDMI {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //1.创建kafkaAdminClient
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.72.15:9092");
        KafkaAdminClient adminClient = (KafkaAdminClient) KafkaAdminClient.create(props);

//        delete(adminClient);
        create(adminClient);
        list(adminClient);
        adminClient.close();
    }

    //删除topic
    private static void delete(KafkaAdminClient adminClient) throws InterruptedException, ExecutionException {
        DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Arrays.asList("topic01"));
        deleteTopicsResult.all().get();  //同步删除
    }

    //查看topic列表
    private static void list(KafkaAdminClient adminClient) throws InterruptedException, ExecutionException {
        ListTopicsResult listTopicsResult = adminClient.listTopics();
        Set<String> names = listTopicsResult.names().get();
        System.out.println("---topic了列表----");
        names.forEach( name -> {
            System.out.println(name);
        });
    }

    //创建topic
    private static void create(KafkaAdminClient adminClient) throws InterruptedException, ExecutionException {
        CreateTopicsResult createResults = adminClient.createTopics(Arrays.asList(new NewTopic("topic01", 3, (short) 1)));
        createResults.all().get(); //同步创建topic
    }
}
